package com.gitee.ywj1352.reactive;

import org.reactivestreams.Subscription;
import reactor.core.Exceptions;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.SignalType;

public class BusinessSubscriber extends BaseSubscriber<Integer> {

    private int max;

    Subscription subscription;

    public int getMax() {
        return max;
    }

    public BusinessSubscriber(int max) {
        this.max = max;
    }

    public void hookOnSubscribe(Subscription subscription) {
        System.out.println("BusinessSubscriber#hookOnSubscribe : "+subscription.getClass());
        this.subscription = subscription;
        subscription.request(max);
    }

    public void hookOnNext(Integer value) {
        System.out.println("do on next " + value);
        if (value == 2){
            subscription.cancel();
        }
    }


    public void hookOnComplete() {
        System.out.println("do on complete");
    }


    public void hookOnError(Throwable throwable) {
        //
        throw Exceptions.errorCallbackNotImplemented(throwable);
    }


    public void hookOnCancel() {
        System.out.println("Cancel");
    }

    public void hookFinally(SignalType type) {
        System.out.println(type);
    }

}
